package com.neusoft.bigdata.mysql.cluster;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * MySQL Cluster集群数据基本操作，以及数据分片功能、性能测试类
 * 
 * @author lihzh
 * @date 2013-1-16 下午8:25:43
 */
public class MySQLClusterDataMachine {

	private static final String DB_IPADDRESS = "10.4.44.200";
	private static final String DB_IPADDRESS_TWO = "10.4.44.201";
	private static final String DB_IPADDRESS_AMEOBA = "10.4.44.205";
	private static final String DB_IPADDRESS_MYSQL = "10.4.44.206";
	private static final int DB_PORT = 3306;
	private static final int AMEOBA_PORT = 8066;
	private static final String DB_NAME = "bigdata";
	private static final String DB_USER = "root";
	private static final String DB_PASSOWRD = "root";

	private static final long BATCH_ROW_COUNT = 100;
	private static final long TOTAL_ROW_COUNT = BATCH_ROW_COUNT * 10;

	private static final String TABLE_NAME_BIGDATACLUSTER = "bigdata_cluster";
	private static final String TABLE_NAME_DATAHOUSE = "data_house";
	private static final String TABLE_NAME_D = "data_house";

	static {
		try {
			Class.forName("com.mysql.jdbc.Driver");
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
	}

	/**
	 * @param args
	 * @author lihzh
	 * @throws SQLException
	 * @throws InterruptedException 
	 * @date 2013-1-16 下午8:25:43
	 */
	public static void main(String[] args) throws SQLException, InterruptedException {
		Connection conn = getConnection(DB_IPADDRESS_MYSQL, DB_PORT, DB_NAME,
				DB_USER, DB_PASSOWRD);
		MySQLClusterDataMachine dataMachine = new MySQLClusterDataMachine();
//		long start = System.currentTimeMillis();
		// 批量向数据库插入数据，5w flush一次。
//		 dataMachine.insertDataToTable(conn, TABLE_NAME, TOTAL_ROW_COUNT,
		// BATCH_ROW_COUNT);
		// 从数据库中查询数据
		// dataMachine.searchData(conn, TABLE_NAME_BIGDATACLUSTER);
		// 新建表
		// dataMachine.createTable(conn, TABLE_NAME_DATAHOUSE);
		// 插入数据
//		dataMachine.insertDataToTable(conn, TABLE_NAME_DATAHOUSE,
//		 TOTAL_ROW_COUNT, BATCH_ROW_COUNT, 0);
		dataMachine.parallelInsert();
//		long end = System.currentTimeMillis();
//		System.out.println(end - start);
//		dataMachine.deleteAll(conn, TABLE_NAME_DATAHOUSE);
		// for (int i = 0; i < 10000; i++) {
		// dataMachine.selectAll(conn, TABLE_NAME_DATAHOUSE);
		// }
	}

	private void selectAll(Connection conn, String tableName)
			throws SQLException {
		long start = System.currentTimeMillis();
		String sql = "SELECT * from " + tableName;
		ResultSet resultSet = conn.createStatement().executeQuery(sql);
		printResult(resultSet);
		long end = System.currentTimeMillis();
		System.out.println("Cost time: " + (end - start));
		resultSet.close();
	}

	private void printResult(ResultSet resultSet) throws SQLException {
		int count = 0;
		while (resultSet.next()) {
			System.out.println(resultSet.getString("name"));
			count++;
		}
		System.out.println("Total count: " + count);
	}

	/**
	 * 创建表，写死包括三个字段，id,name,description。
	 * 
	 * @param conn
	 * @param tableNameDatahouse
	 * @author lihzh
	 * @throws SQLException
	 * @date 2013-1-17 下午5:49:33
	 */
	private void createTable(Connection conn, String tableName)
			throws SQLException {
		String sql = "CREATE table "
				+ tableName
				+ " (id int primary key, name varchar(20), description varchar(20)) engine=ndbcluster default charset utf8;";
		conn.createStatement().execute(sql);
		System.out.println("Create table: " + tableName + " finished.");
	}

	/**
	 * 查询特定的一个或者一组数据，打印查询耗时
	 * 
	 * @param conn
	 * @param tableName
	 * @author lihzh
	 * @throws SQLException
	 * @date 2013-1-17 下午4:46:20
	 */
	private void searchData(Connection conn, String tableName)
			throws SQLException {
		long start = System.currentTimeMillis();
		String sql = "select * from "
				+ tableName
				+ " where id in (85902, 123,123445,30945,234,456456,23412,4093423,39420,999999)";
		PreparedStatement statement = conn.prepareStatement(sql);
		ResultSet resultSet = statement.executeQuery();
		resultSet.last();
		System.out.println("Name is: " + resultSet.getObject("name"));
		long end = System.currentTimeMillis();
		System.out.println("Query ten rows from table [" + tableName
				+ "] cost time: " + (end - start));
	}

	/**
	 * 多线程并行写入测试
	 * 
	 * @author lihzh(OneCoder)
	 * @blog http://www.coderli.com
	 * @date 2013-2-27 下午3:39:56
	 */
	private void parallelInsert() {
		final long start = System.currentTimeMillis();
		Thread t1 = new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					Connection conn = getConnection(DB_IPADDRESS_MYSQL, DB_PORT,
							DB_NAME, DB_USER, DB_PASSOWRD);
					MySQLClusterDataMachine dataMachine = new MySQLClusterDataMachine();
					dataMachine.insertDataToTable(conn, TABLE_NAME_DATAHOUSE,
							500, 100, 0);
					long end1 = System.currentTimeMillis();
					System.out.println("Thread 1 cost: " + (end1 - start));
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
		});

		Thread t2 = new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					Connection conn = getConnection(DB_IPADDRESS_MYSQL, DB_PORT,
							DB_NAME, DB_USER, DB_PASSOWRD);
					MySQLClusterDataMachine dataMachine = new MySQLClusterDataMachine();
					dataMachine.insertDataToTable(conn, TABLE_NAME_DATAHOUSE,
							500, 100, 500);
					long end2 = System.currentTimeMillis();
					System.out.println("Thread 2 cost: " + (end2 - start));
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
		});
		Thread t3 = new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					Connection conn = getConnection(DB_IPADDRESS_MYSQL, DB_PORT,
							DB_NAME, DB_USER, DB_PASSOWRD);
					MySQLClusterDataMachine dataMachine = new MySQLClusterDataMachine();
					dataMachine.insertDataToTable(conn, TABLE_NAME_DATAHOUSE,
							500, 100, 1000);
					long end2 = System.currentTimeMillis();
					System.out.println("Thread 3 cost: " + (end2 - start));
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
		});
		t1.start();
		t2.start();
		t3.start();
	}

	/**
	 * 向数据库中插入数据
	 * 
	 * @param conn
	 * @param totalRowCount
	 * @param perRowCount
	 * @param tableName
	 * @author lihzh(OneCoder)
	 * @throws SQLException
	 * @date 2013-1-17 下午1:57:10
	 */
	private void insertDataToTable(Connection conn, String tableName,
			long totalRowCount, long perRowCount, long startIndex)
			throws SQLException {
		String sql = "insert into " + tableName + " VALUES(?,?,?)";
		System.out.println("Begin to prepare statement.");
		PreparedStatement statement = conn.prepareStatement(sql);
		long sum = 0L;
		for (int j = 0; j < totalRowCount / perRowCount; j++) {
			long batchStart = System.currentTimeMillis();
			for (int i = 0; i < perRowCount; i++) {
				long id = j * perRowCount + i + startIndex;
				String name_pre = String.valueOf(id);
				statement.setLong(1, id);
				statement.setString(2, name_pre);
				statement.setString(3, name_pre);
				statement.addBatch();
			}
			System.out.println("It's up to batch count: " + perRowCount);
			statement.executeBatch();
			conn.commit();
			long batchEnd = System.currentTimeMillis();
			long cost = batchEnd - batchStart;
			System.out
					.println("Batch data commit finished. Time cost: " + cost);
			sum += cost;
		}
		System.out.println("All data insert finished. Total time cost: " + sum);
		System.out.println("Avg cost: " + sum / 10);
	}

	/**
	 * 清空表中所有记录
	 * 
	 * @param conn
	 * @param tableName
	 * @author lihzh
	 * @throws SQLException
	 * @date 2013-2-27 上午10:18:15
	 */
	private void deleteAll(Connection conn, String tableName)
			throws SQLException {
		String sql = "DELETE FROM " + tableName;
		conn.createStatement().execute(sql);
		conn.commit();
		System.out.println("delete " + tableName + " data finished.");
	}

	/**
	 * 获取MySQL数据库链接
	 * 
	 * @param ip
	 * @param port
	 * @param db
	 * @param username
	 * @param password
	 * @return
	 * @throws SQLException
	 * @author lihzh
	 * @date 2013-1-17 上午11:44:10
	 */
	private static Connection getConnection(String ip, int port, String db,
			String username, String password) throws SQLException {
		StringBuilder strBuilder = new StringBuilder();
		String connUrl = strBuilder.append("jdbc:mysql://").append(ip)
				.append(":").append(port).append("/").append(db).toString();
		Connection conn = DriverManager.getConnection(connUrl, username,
				password);
		conn.setAutoCommit(false);
		return conn;
	}
}
